有 Java 编程相关的问题?

你可以在下面搜索框中键入要查询的问题!

java如何使用IDE在storm production集群中提交拓扑

在使用IDE向生产集群提交拓扑时,我面临一个问题Must submit topologies using the 'storm' client script so that StormSubmitter knows which jar to upload,而如果我在命令行中使用storm jar命令执行同样的操作,它的运行就像天堂一样。我在githublink中看到过同样的例子

对于提交拓扑,我使用这些行集

conf.put(Config.NIMBUS_HOST, NIMBUS_NODE);
conf.put(Config.NIMBUS_THRIFT_PORT,6627);
conf.put(Config.STORM_ZOOKEEPER_PORT,2181);
conf.put(Config.STORM_ZOOKEEPER_SERVERS,ZOOKEEPER_ID);
conf.setNumWorkers(20);
conf.setMaxSpoutPending(5000);
StormSubmitter submitter = new StormSubmitter();
submitter.submitTopology("test", conf, builder.createTopology());

请告诉我这是否是正确的运行方法


共 (4) 个答案

  1. # 2 楼答案

    我们找到了解决办法。当我们运行“storm jar”时,它会触发storm的属性标志。提交的jar中的jar。因此,如果我们想以编程方式提交一个jar,那么只需这样设置标志

    System.setProperty("storm.jar", <path-to-jar>);

    例如:

    System.setProperty("storm.jar", "/Users/programming/apache-storm-1.0.1/lib/storm-core-1.0.1.jar");
    StormSubmitter.submitTopology("myTopology", config, builder.createTopology());
    
  2. # 3 楼答案

    我已经根据@abhi和@Nishu Tayal的答案解决了这个问题,我想在这里发布我的代码:

    public static void submitLocalTopologyWay1(String topologyName, Config topologyConf, 
            StormTopology topology, String localJar) {
        try {
            //get default storm config
            Map defaultStormConf = Utils.readStormConfig();
            defaultStormConf.putAll(topologyConf);
    
            //set JAR
            System.setProperty("storm.jar",localJar);
    
            //submit topology
            StormSubmitter.submitTopology(topologyName, defaultStormConf, topology);
    
        } catch (Exception e) {
            String errorMsg = "can't deploy topology " + topologyName + ", " + e.getMessage();
            System.out.println(errorMsg);
            e.printStackTrace();
        } 
    }
    
    public static void submitLocalTopologyWay2(String topologyName, Config topologyConf, 
            StormTopology topology, String localJar) {
        try {
            //get nimbus client
            Map defaultStormConf = Utils.readStormConfig();
            defaultStormConf.putAll(topologyConf);
            Client client = NimbusClient.getConfiguredClient(defaultStormConf).getClient();
    
            //upload JAR
            String remoteJar = StormSubmitter.submitJar(defaultStormConf, localJar);
    
            //submit topology
            client.submitTopology(topologyName, remoteJar, JSONValue.toJSONString(topologyConf), topology);
    
        } catch (Exception e) {
            String errorMsg = "can't deploy topology " + topologyName + ", " + e.getMessage();
            System.out.println(errorMsg);
            e.printStackTrace();
        } 
    }
    

    接下来是一个测试,您必须首先将代码构建到JAR文件中

    public void testSubmitTopologySubmitLocalTopologyWay1() {   
        Config config = new Config();
        config.put(Config.NIMBUS_HOST,"9.119.84.179");   
        config.put(Config.NIMBUS_THRIFT_PORT, 6627);
        config.put(Config.STORM_ZOOKEEPER_SERVERS, Arrays.asList("9.119.84.177","9.119.84.178","9.119.84.176")); 
        config.put(Config.STORM_ZOOKEEPER_PORT,2181);
    
        config.put(Config.TOPOLOGY_WORKERS, 3);
    
        RemoteSubmitter.submitLocalTopologyWay1("word-count-test-1", config, 
                WordCountTopology.buildTopology(), // your topology
                "C:\\MyWorkspace\\project\\storm-sample-0.0.1-SNAPSHOT-jar-with-dependencies.jar");//the JAR file
    }
    
  3. # 4 楼答案

    为了向远程Storm集群提交拓扑,您需要将该jar上传到nimbus机器,然后使用NimbusClient将该jar提交到集群
    您可以这样做:

    Map storm_conf = Utils.readStormConfig();
    storm_conf.put("nimbus.host", "<Nimbus Machine IP>");
    Client client = NimbusClient.getConfiguredClient(storm_conf)
                                    .getClient();
    String inputJar = "C:\\workspace\\TestStormRunner\\target\\TestStormRunner-0.0.1-SNAPSHOT-jar-with-dependencies.jar";
    NimbusClient nimbus = new NimbusClient(storm_conf, "<Nimbus Machine IP>",
                                    <Nimbus Machine Port>);
     // upload topology jar to Cluster using StormSubmitter
    String uploadedJarLocation = StormSubmitter.submitJar(storm_conf,
                                    inputJar);
    
    String jsonConf = JSONValue.toJSONString(storm_conf);
    nimbus.getClient().submitTopology("testtopology",
                          <uploadedJarLocation>, jsonConf, builder.createTopology());
    

    下面是一个工作示例:Submitting a topology to Remote Storm Cluster